package com.alison.tableapisql.util;

import org.apache.calcite.avatica.util.Casing;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.parser.SqlParser;
import org.apache.flink.sql.parser.impl.FlinkSqlParserImpl;
import org.apache.flink.sql.parser.validate.FlinkSqlConformance;

import java.util.ArrayList;
import java.util.List;

import static org.apache.calcite.avatica.util.Quoting.BACK_TICK;

/**
 * @Author alison
 * @Date 2024/4/12 17:48
 * @Version 1.0
 * @Description
 */
public class FlinkSqlParseUtil {


    public static void main(String[] args) {
        String string = "create table dataTable (" +
                " id string not null, " +
                " ts bigint, " +
                " temperature double, " +
                " pt AS PROCTIME() " +
                " ) with (" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'file:///sensor.txt', " +
                " 'format' = 'csv')";
        string = "create table dataTable (" +
                " id string not null, " +
                " ts bigint, " +
                " temperature double, " +
                " rt AS TO_TIMESTAMP( FROM_UNIXTIME(ts) ), " +
                " watermark for rt as rt - interval '1' second" +
                " ) with (" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'file:///sensor.txt', " +
                " 'format' = 'csv')";
        string = "select id, top2(temp) as (tempv, rank)" +
                " from sensor group by id ";
        parseFlinkSql(string);
    }

    public static List<String> parseFlinkSql(String sql) {
        List<String> sqlList = new ArrayList<>();
        if (sql != null && !sql.isEmpty()) {
            try {
                SqlParser parser = SqlParser.create(sql, SqlParser.configBuilder()
                        .setParserFactory(FlinkSqlParserImpl.FACTORY)
                        .setQuoting(BACK_TICK)
                        .setUnquotedCasing(Casing.TO_LOWER)   //字段名统一转化为小写
                        .setQuotedCasing(Casing.UNCHANGED)
                        .setConformance(FlinkSqlConformance.DEFAULT)
                        .build()
                );
                List<SqlNode> sqlNodeList = parser.parseStmtList().getList();
                if (sqlNodeList != null && !sqlNodeList.isEmpty()) {
                    for (SqlNode sqlNode : sqlNodeList) {
                        sqlList.add(sqlNode.toString());
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return sqlList;
    }


    private FlinkSqlParseUtil() {
    }

}
